package net.pws.common.simplemq;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DefaultDataQueueProcessor implements DataQueueProcessor, DataPipe {
    
    private ExecutorService executorService;
    
    private ScheduledExecutorService scheduledExecutorService;
    
    private Action action;
    
    public DefaultDataQueueProcessor(Action action) {
        this.action = action;
        this.executorService = Executors.newFixedThreadPool(5);
        this.scheduledExecutorService = Executors.newScheduledThreadPool(5);
    }
    
    public DefaultDataQueueProcessor(Action action, int handlerSize) {
        this.action = action;
        this.executorService = Executors.newFixedThreadPool(handlerSize);
        this.scheduledExecutorService = Executors.newScheduledThreadPool(handlerSize);
    }
    
    public void stop() {
        if (!executorService.isShutdown()) {
            executorService.shutdown();
        }
    }
    
    public DataPipe getDataPipe() {
        return this;
    }
    
    public void addData(final Object data) {
        executorService.execute(new Runnable() {
            
            public void run() {
                action.execute(new Context() {
                    
                    public Object getData() {
                        return data;
                    }
                    
                    public DataPipe getQueue() {
                        return DefaultDataQueueProcessor.this;
                    }
                    
                });
            }
            
        });
    }
    
    public void addLazyData(final Object data) {
        addLazyData(data, 60 * 5);
    }
    
    public void addLazyData(final Object data, int delay) {
        scheduledExecutorService.schedule(new Runnable() {
            
            public void run() {
                action.execute(new Context() {
                    
                    public Object getData() {
                        return data;
                    }
                    
                    public DataPipe getQueue() {
                        return DefaultDataQueueProcessor.this;
                    }
                    
                });
            }
            
        }, delay, TimeUnit.SECONDS);
    }
    
}
